/*
 * Copyright 2021 TiKV Project Authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package com.webank.tikv.region;

import com.google.protobuf.ByteString;
import com.webank.tikv.TiConfiguration;
import com.webank.tikv.exception.InternalException;
import com.webank.tikv.kvproto.Kvrpcpb;
import com.webank.tikv.kvproto.Metapb;
import com.webank.tikv.util.FastByteComparisons;

import java.io.Serializable;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public class TiRegion implements Serializable {

    private final Kvrpcpb.APIVersion API_V1 = Kvrpcpb.APIVersion.V1;

    private final Metapb.Region meta;
    private final Kvrpcpb.IsolationLevel isolationLevel;
    private final Kvrpcpb.CommandPri commandPri;
    private final TiConfiguration conf;
    private final Metapb.Peer leader;
    private final List<Metapb.Peer> replicas;
    private final AtomicInteger replicaIdx;
    private final List<TiStore> stores;
    private final ReplicaSelector replicaSelector;
    private final String resource;

    public TiRegion(TiConfiguration conf, Metapb.Region meta, Metapb.Peer leader,
                    List<Metapb.Peer> peers, List<TiStore> stores, ReplicaSelector replicaSelector) {
        this.conf = Objects.requireNonNull(conf, "conf is null");
        this.meta = Objects.requireNonNull(meta, "meta is null");
        if (leader == null || leader.getId() == 0) {
            throw new InternalException("leader is null for region:" + meta.getId());
        }
        this.isolationLevel = conf.getIsolationLevel();
        this.commandPri = conf.getCommandPriority();
        this.replicaSelector = replicaSelector;
        this.stores = stores;
        this.leader = leader;
        this.resource = conf.getResource();

        replicas = replicaSelector.select(this.leader, peers, stores);
        replicaIdx = new AtomicInteger();
    }

    public TiConfiguration getConf() {
        return conf;
    }

    public Metapb.Peer getLeader() {
        return leader;
    }

    public List<Metapb.Peer> getFollowers() {
        return meta.getPeersList().stream()
            .filter(peer -> !peer.equals(leader) && peer.getRole().equals(Metapb.PeerRole.Voter))
            .collect(Collectors.toList());
    }

    public Metapb.Peer getNextReplica() {
        int idx = replicaIdx.getAndIncrement();
        if (idx == Integer.MAX_VALUE) {
            idx = 0;
            replicaIdx.set(0);
        }
        return replicas.get(idx % replicas.size());
    }

    private boolean isLeader(Metapb.Peer peer) {
        return leader.equals(peer);
    }

    public long getId() {
        return meta.getId();
    }

    public ByteString getStartKey() {
        return meta.getStartKey();
    }

    public ByteString getEndKey() {
        return meta.getEndKey();
    }

    public Kvrpcpb.Context makeLeaderContext() {
        return makeContext(leader, false);
    }

    public Kvrpcpb.Context makeReplicaContext(Metapb.Peer peer) {
        return makeContext(peer, true);
    }

    private Kvrpcpb.Context makeContext(Metapb.Peer peer, boolean replicaRead) {
        return Kvrpcpb.Context.newBuilder()
            .setApiVersion(API_V1)
            .setRegionId(meta.getId())
            .setRegionEpoch(meta.getRegionEpoch())
            .setPeer(peer)
            .setReplicaRead(replicaRead)
            .setPriority(commandPri)
            .setIsolationLevel(isolationLevel)
            .setRequestSource(resource)
            .build();
    }

    // getVerID returns the Region's RegionVerID.
    public RegionVerID getVerID() {
        return new RegionVerID(
            meta.getId(), meta.getRegionEpoch().getConfVer(), meta.getRegionEpoch().getVersion());
    }

    /**
     * switches current peer to the one on specific store. It return false if no peer matches the
     * storeID.
     *
     * @param leaderStoreID is leader peer id.
     * @return null if no peers matches the store id.
     */
    public TiRegion switchPeer(long leaderStoreID) {
        List<Metapb.Peer> peers = meta.getPeersList();
        for (Metapb.Peer p : peers) {
            if (p.getStoreId() == leaderStoreID) {
                return new TiRegion(conf, meta, p, peers, stores, replicaSelector);
            }
        }
        return null;
    }

    public boolean isMoreThan(ByteString key) {
        return FastByteComparisons.compareTo(
            meta.getStartKey().toByteArray(), 0, meta.getStartKey().size(),
            key.toByteArray(), 0, key.size()) > 0;
    }

    public boolean isLessThan(ByteString key) {
        return FastByteComparisons.compareTo(
            meta.getEndKey().toByteArray(), 0, meta.getEndKey().size(),
            key.toByteArray(), 0, key.size()) <= 0;
    }

    public boolean contains(ByteString key) {
        return !isMoreThan(key) && !isLessThan(key);
    }

    public boolean isValid() {
        return leader != null && meta != null;
    }

    public Metapb.RegionEpoch getRegionEpoch() {
        return meta.getRegionEpoch();
    }

    public Metapb.Region getMeta() {
        return meta;
    }

    @Override
    public boolean equals(final Object another) {
        if (!(another instanceof TiRegion)) {
            return false;
        }
        TiRegion anotherRegion = ((TiRegion) another);
        return anotherRegion.meta.equals(this.meta)
            && anotherRegion.leader.equals(this.leader)
            && anotherRegion.commandPri.equals(this.commandPri)
            && anotherRegion.isolationLevel.equals(this.isolationLevel);
    }

    @Override
    public int hashCode() {
        return Objects.hash(meta, leader, isolationLevel, commandPri);
    }

    @Override
    public String toString() {
        Metapb.RegionEpoch regionEpoch = meta.getRegionEpoch();
        return String.format("[Region:%d, ConfVer:%d, Version:%d, Leader:%d, Store:%d]",
            getId(), regionEpoch.getConfVer(), regionEpoch.getVersion(), leader.getId(), leader.getStoreId());
    }

    public static class RegionVerID {

        final long id;
        final long confVer;
        final long ver;

        RegionVerID(long id, long confVer, long ver) {
            this.id = id;
            this.confVer = confVer;
            this.ver = ver;
        }

        public long getId() {
            return id;
        }

        public long getConfVer() {
            return confVer;
        }

        public long getVer() {
            return ver;
        }

        @Override
        public boolean equals(Object other) {
            if (this == other) {
                return true;
            }
            if (!(other instanceof RegionVerID)) {
                return false;
            }

            RegionVerID that = (RegionVerID) other;
            return id == that.id && confVer == that.confVer && ver == that.ver;
        }

        @Override
        public int hashCode() {
            int hash = Long.hashCode(id);
            hash = hash * 31 + Long.hashCode(confVer);
            hash = hash * 31 + Long.hashCode(ver);
            return hash;
        }
    }
}
